package com.example.mqtt.mq;

import com.alibaba.fastjson2.JSONObject;
import com.example.mqtt.config.MessageQueueManager;
import com.example.mqtt.config.MessageStorage;
import com.example.mqtt.config.MqttConfig;
import com.example.mqtt.entity.QueueInfo;
import com.example.mqtt.entity.User;
import com.example.mqtt.model.Card;
import com.example.mqtt.service.QueueService;
import com.example.mqtt.service.WebSocketService;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author qwert
 * @version 1.0.0
 * @create 2025/4/21 21:44
 **/
@Slf4j
@Component
public class MqttSubscriberAddQueue {

    private static final String TOPIC = "sensor/addqueue";

    @Autowired
    private MqttConfig mqttConfig;
    @Autowired
    private QueueService queueService;

    @PostConstruct
    public void subscribe() {
        Mqtt5AsyncClient mqttClient = mqttConfig.mqttClient();
        mqttClient.subscribeWith()
                .topicFilter(TOPIC)
                .qos(MqttQos.AT_LEAST_ONCE)
                .callback(this::handleMessage)
                .send()
                .whenComplete((subAck, throwable) -> {
                    if (throwable != null) {
                        log.warn("订阅失败: " + throwable.getMessage());
                    } else {
                        log.info("成功订阅主题: " + TOPIC);
                    }
                });
    }

    private void handleMessage(Mqtt5Publish publish) {
        String payload = new String(publish.getPayloadAsBytes());
        log.info("收到消息 [主题: " + publish.getTopic() + "]: " + payload);
        QueueInfo queueInfo = JSONObject.parseObject(payload, QueueInfo.class);
        queueService.pushQueue(queueInfo);
    }
}

